package com.qingcloud.df.executor.consumer;

import cn.hutool.extra.spring.SpringUtil;
import com.qingcloud.df.executor.cluster.ServerService;
import com.qingcloud.df.sdk.component.message.NodeInfo;
import lombok.extern.slf4j.Slf4j;
import java.util.List;

@Slf4j
public class HeartbeatConsumer extends BaseConsumer<NodeInfo> {

    public HeartbeatConsumer(Long id,
                             Integer consumerCoreThreadNum,
                             Integer consumerMaxThreadNum,
                             String consumerThreadName,
                             Integer pollSize,
                             String pollThreadName,
                             TaskQueue<NodeInfo> queues) {
        super(id, consumerCoreThreadNum, consumerMaxThreadNum, consumerThreadName, pollSize, pollThreadName, queues, 2000L, 1000L);
    }

    @Override
    public void consume(Long id, List<NodeInfo> nodeInfos) {
        this.consumerExecutor.submit(new ExecutorHeartbeatConsumerRunnable(nodeInfos));
    }

    private record ExecutorHeartbeatConsumerRunnable(List<NodeInfo> nodeInfos) implements Runnable {
        @Override
            public void run() {
                try {
                    SpringUtil.getBean(ServerService.class).batchHeartbeat(nodeInfos);
                } catch (Throwable throwable) {
                    log.error("Executor heartbeat failed!", throwable);
                }
            }
        }
}
